import java.io.IOException;
import java.util.*;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;

/**
 * This is a very simple Reducer that combines [src, dest] to [src, [dest
 * list]] format.
 *
 * @author Ian Voysey
 * @version 0.1
 */
public class JesusReducer extends MapReduceBase
    implements Reducer<Text, Text, Text, Text> {

    private String searchKey = "Jesus";

    public void reduce(Text key,
		       Iterator<Text> values,
		       OutputCollector<Text, Text> output,
		       Reporter reporter)
	throws IOException {

        // Dist is the distance from the searchkey to this key
        int Dist = Integer.MAX_VALUE;
        if(key.toString().equals(searchKey))
            Dist = 0;

        HashSet<String> linkSet = new HashSet<String>();

	while (values.hasNext()) {
            String next = "";
            String[] nextArr = values.next().toString().split("\\t");

            // If a combiner has already been through here, just grab the links
            // Note: nextArr[0] is dist if length > 1, otherwise, it is the links
            if(nextArr.length > 1){
                next = nextArr[1];
	    } else {
                next = nextArr[0];
	    }

	    // ignore empty links; these are isolated nodes
            if(next != "")
                linkSet.add(next);
	}

        // Links are concated and separated by semicolons
        String links = "";
        for(String link : linkSet)
            links += link + ";";

        int length = links.length();
        if(length <= 0)
            output.collect(key, new Text(Dist+"\t;"));
        else
	    output.collect(key, new Text(Dist+"\t"+links));
    }
}
